package com.moli.iov.handle.worker.his.lmax;

import com.moli.iov.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 *
 * Disruptor service of coroutine pool.
 *
 * <p>
 *  According group to Load balancing every one of coroutine.
 * <p/>
 *
 * @param <T> Commit to data.
 * @param <R> Disruptor task.
 *
 * @author OAK
 * @since 2020/01/16 17:25 PM.
 * @version 3.0.1
 *
 */
public class DisruptorService<T extends ConcurrentMap, R extends DisruptorTask> implements Disruptor<T, R> {

    protected static final Logger logger = LoggerFactory.getLogger(DisruptorService.class);

    /**
     * Default coroutine size.
     */
    private static final int DEFAULT_COROUTINE_NUMBERS = Runtime.getRuntime().availableProcessors();

    /**
     * Default async coroutine size.
     */
    private static final int DEFAULT_ASYNC_COROUTINE_SIZE = 500;

    /**
     * Default delay time.
     */
    private static final int DEFAULT_DELAY_TIME = 500;

    /**
     * Max coroutine size.
     */
    private static final int MAX_COROUTINE_NUMBERS = DEFAULT_COROUTINE_NUMBERS * 50;

    /**
     * Min coroutine size.
     */
    private static final int MIN_COROUTINE_NUMBERS = 1;

    /**
     * Async coroutine size.
     */
    private int asyncCoroutineSize;

    /**
     * Coroutine pool size.
     */
    private int coroutineSize;

    /**
     * Queue size;
     */
    private int queueSize = 1024;

    /**
     * Over time.
     */
    private int delayTime;

    /**
     * Coroutine collection.
     */
    private final List<Coroutine> coroutines = Collections.synchronizedList(new ArrayList<Coroutine>());

    /**
     * According num to statics a coroutine name of suffix.
     */
    private AtomicInteger coroutineNum = new AtomicInteger();

    /**
     * According default coroutine number and async coroutine size to initialize coroutine pool.
     */
    public DisruptorService() {
        this.coroutineSize = DEFAULT_COROUTINE_NUMBERS;
        this.asyncCoroutineSize = DEFAULT_ASYNC_COROUTINE_SIZE;
        this.delayTime = DEFAULT_DELAY_TIME;
    }

    /**
     * Coroutine name of prefix.
     */
    private String coroutineName = "Disruptor-Coroutine-";

    /**
     * According coroutine num and queue size and thread name
     * @param coroutineNum Coroutine num.
     * @param queueSize Queue size.
     * @param coroutineName Coroutine name.
     * @param asyncCoroutineSize Async coroutine size.
     * @param delayTime Delay time.
     */
    public DisruptorService(int coroutineNum, int queueSize, String coroutineName, int asyncCoroutineSize, int delayTime) {
        this.queueSize = queueSize;
        this.coroutineName = coroutineName;
        this.asyncCoroutineSize = asyncCoroutineSize;
        coroutineNum = coroutineNum > MAX_COROUTINE_NUMBERS ? MAX_COROUTINE_NUMBERS : coroutineNum;
        coroutineNum = coroutineNum < MIN_COROUTINE_NUMBERS ? MIN_COROUTINE_NUMBERS : coroutineNum;
        this.coroutineSize = coroutineNum;
        this.delayTime = delayTime;
    }

    /**
     * According <R> Disruptor task to Initializing every one of disruptor batch thread pool.
     * @param task <R> Disruptor task.
     * @return Disruptor batch thread pool.
     */
    @Override
    public Disruptor<T, R> disruptor(R task) {
        coroutines.stream().forEach(coroutine -> coroutine.setTask(task));
        return this;
    }

    /**
     * Publish disruptor batch thread pool current group of <T> commit to data.
     * @param group Current group.
     * @param data <T> Commit to data.
     * @throws InterruptedException  Interrupted exception @link{InterruptedException}.
     */
    @Override
    public void publish(String group, T data) throws InterruptedException {
        getCoroutine(group).publish(data);
    }

    /**
     * According group to get Coroutine.
     * @param group Current coroutine group.
     * @return a coroutine.
     */
    public Coroutine getCoroutine(String group){
        if(StringUtil.isBlank(group)){
            throw new NullPointerException("Current coroutine group is null.");
        }
        return coroutines.get(getIndex(group));
    }

    /**
     * Get the few of group.
     * @param group Current coroutine group.
     * @return
     */
    public int getIndex(String group){
        return Math.abs(group.hashCode()) % coroutines.size();
    }

    /**
     * Shut down all coroutine.
     */
    @Override
    public void shutdown() {
        for(Coroutine coroutine : coroutines){
            coroutine.shutdown();
        }
    }

    /**
     * Get coroutine size.
     * @return a coroutine size.
     */
    @Override
    public int getCoroutineSize() {
        return coroutines.size();
    }

    /**
     * Set queue size.
     * @param queueSize Queue size.
     */
    public void setQueueSize(int queueSize) {
        this.queueSize = queueSize;
    }

    /**
     * Set coroutine Name.
     * @param coroutineName Coroutine name.
     */
    public void SetCoroutineName(String coroutineName) {
        this.coroutineName = coroutineName;
    }

    /**
     * According initialize coroutine num and async coroutine size to initialize coroutine.
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        for(int i = 0; i < this.coroutineSize; i++){
            Coroutine coroutine = new Coroutine(asyncCoroutineSize, delayTime);
            coroutines.add(coroutine);
            String coroutineAliasName = String.format("%s%d", this.coroutineName, coroutineNum.incrementAndGet());
            Thread thread = new Thread(coroutine, coroutineAliasName);
            thread.start();
            logger.info("[Coroutine pool] initialize [{}] coroutine", coroutineAliasName);
        }
    }

    /**
     * Coroutine.
     */
    public class Coroutine implements Runnable {

        private BlockingQueue<T> blockingQueue;

        private volatile Integer asyncCoroutineSize;

        private volatile Integer delayTime;

        private volatile R task;

        private volatile Instant delay;

        public void setTask(R task){
            this.task = task;
        }

        /**
         * Whether to current disruptor thread pool of running.
         */
        private volatile boolean running = true;

        public Coroutine() {
            logger.info("initialize blockingQueue size: {}", queueSize);
            this.blockingQueue =  new LinkedBlockingQueue(queueSize);
            this.asyncCoroutineSize = DEFAULT_ASYNC_COROUTINE_SIZE;
            this.delayTime = DEFAULT_DELAY_TIME;
            delay = Instant.now();
        }

        public Coroutine(Integer asyncCoroutineSize, Integer delayTime) {
            this();
            this.asyncCoroutineSize = asyncCoroutineSize;
            this.delayTime = delayTime;
        }


        /**
         * Publish current <T> data to Blocking queue.
         * @param data <T> data.
         * @throws InterruptedException
         */
        public void publish(T data) throws InterruptedException {
            try {
                this.blockingQueue.put(data);
            } catch (InterruptedException e) {
                logger.error("Coroutine [{}] publish Error:{}",  getClass(), e.getMessage(), e);
                throw e;
            }
        }

        @Override
        public void run() {
            while(running){
                ConcurrentMap<String, T> concurrentMap = new ConcurrentHashMap<>(asyncCoroutineSize);
                for(int i = 0; i < asyncCoroutineSize; i++){
                    T data = null;
                    try {
                        data = this.blockingQueue.take();
                    } catch (Exception e) {
                        logger.error("Coroutine [{}] take Error:{}",  getClass(), e.getMessage(), e);
                    }
                    try {
                        if(data != null){
                            concurrentMap.putAll(data);
                        }
                    } catch (Exception e) {
                        logger.error("Coroutine [{}] put all data Error:{}",  getClass(), e.getMessage(), e);
                    }
                }
                try {
                    if(concurrentMap.size() == asyncCoroutineSize){
                        task.subscribe(concurrentMap);
                        delay = Instant.now();
                    }else{
                        if(delayTime < Instant.now().minusSeconds(delay.getEpochSecond()).getEpochSecond()){
                            task.subscribe(concurrentMap);
                        }
                    }
                } catch (Exception e) {
                    logger.error("Coroutine [{}] run Error:{}",  getClass(), e.getMessage(), e);
                }
            }
        }

        public void shutdown(){
            this.running = false;
        }
    }
}